import com.fasterxml.jackson.databind.{DeserializationFeature, ObjectMapper}
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}

import java.time.LocalDateTime
import java.time.format.DateTimeFormatter
import java.util
import scala.util.Random

/**
 * @author guchenbo
 * @date 2022/4/15
 */
object HolmesKafkaProducer {

  /**
   * 支持scala原生类
   */
  private lazy val _defaultMapper = new ObjectMapper()
    .registerModule(com.fasterxml.jackson.module.scala.DefaultScalaModule)

  /**
   * Override this method if you need a custom Jackson object mapper; the default mapper
   * has the default configuration, plus the Scala module.
   */
  protected def createMapper(): ObjectMapper = _defaultMapper

  final val mapper = createMapper()
  mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)

  def getRawConfObj(allConfStr: String): java.util.HashMap[String, Any] = {
    mapper.readValue(allConfStr, classOf[java.util.HashMap[String, Any]])
  }

  def convertToJsonStr(any: Any): String = {
    mapper.writeValueAsString(any)
  }

  def main(args: Array[String]) {

    val brokers = "10.57.16.13:9092"
    val topic = "holmes-api-default-holmes-call-log"


    val props = new util.HashMap[String, Object]()
    props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers)
    props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
      "org.apache.kafka.common.serialization.StringSerializer")
    val producer = new KafkaProducer[String, String](props)

    val temp = "{\"modelEmptyInputs\":[\"age\",\"sex\",\"RevolvingUtilizationOfUnsecuredLines\",\"NumberOfTime30-59DaysPastDueNotWorse\",\"DebtRatio\",\"MonthlyIncome\",\"NumberOfOpenCreditLinesAndLoans\",\"NumberOfTimes90DaysLate\"],\"inputData\":{},\"data\":{\"probability\":[0.7103809797849969,0.289619020215003],\"target\":0},\"batchData\":[],\"inputs\":{\"NumberOfOpenCreditLinesAndLoans\":-999,\"NumberOfTime30-59DaysPastDueNotWorse\":-999,\"RevolvingUtilizationOfUnsecuredLines\":-999,\"DebtRatio\":-999,\"NumberOfTimes90DaysLate\":-999,\"sex\":-999,\"age\":-999,\"MonthlyIncome\":-999},\"modelVersion\":\"V1\",\"diffDatas\":{},\"index\":\"2022082685779634V1\",\"modelType\":\"P\",\"modelUuid\":\"2022061522531913\",\"modelName\":\"liux_lightGBM\",\"callNumber\":\"5fe9014ce40846068bf32ca7b8647244\",\"pmmlInput\":{},\"success\":false,\"testGroup\":false,\"id\":\"1661755219786-90866742\",\"seqId\":\"2022082975782697\",\"modelTestFlag\":\"holmes_web_model_test_flag_@tongdun\",\"spendTime\":130,\"timestamp\":1661755219634}"

    val list = List[Map[String, String]](
      Map[String, String](
        "modelUuid" -> "2022072652237425",
        "modelName" -> "评分卡001"
      ),
      Map[String, String](
        "modelUuid" -> "2022082685779634",
        "modelName" -> "liux_lightGBM"
      )
    )

    def seqId(): String = {
      // yyyy-MM-dd HH:mm:ss.SSS
      val timestamp = DateTimeFormatter.ofPattern(s"yyyyMMddHHmmssSSS").format(LocalDateTime.now())
      timestamp
    }

    def buildLog(i: Int): String = {
      val idx = i % list.length
      val odd = Random.nextInt(10) % 2 == 1
      val m = list(idx)
      val map = getRawConfObj(temp)
      map.put("modelUuid", m("modelUuid"))
      map.put("modelName", m("modelName"))
      map.put("spendTime", Random.nextInt(100))
      map.put("timestamp", System.currentTimeMillis())
      map.put("seqId", seqId())
      map.put("success", odd)

      convertToJsonStr(map)

    }

    // Send some messages
    val start = LocalDateTime.now()
    val end = start.plusMinutes(5)
    var cnt = 0;
    while (true) {
      val size = Random.nextInt(10)
      for (i <- 1 to size) {
        val str = buildLog(i)
        //        println(i)
        //                println(str)
        cnt += 1

        val message = new ProducerRecord[String, String](topic, Random.nextInt(3).toString, str)
        producer.send(message)
        println(s"write msg $cnt")
      }
      if (LocalDateTime.now().isAfter(end)) {
        println("time is end")
        return
      }
    }
  }
}
